package com.huitone.smspfm.socket.core.net;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.huitone.smspfm.socket.core.handler.AbstractMessageHandler;
import com.huitone.smspfm.socket.core.handler.DefaultMessageHandler;
import com.huitone.smspfm.socket.model.AccountInfo;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.IBody;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit2;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit2Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;
import com.huitone.smspfm.socket.utils.SocketUtils;


public class ExtSocket extends Socket {
	
	private static final Logger logger = LogManager.getLogger(ExtSocket.class);
	
	private final Object readLock = new Object();
	private final Object writeLock = new Object();
	
	/**
	 * 帐号信息
	 */
	private AccountInfo accountInfo = null;
	
	/**
	 * 是否循环读取信息
	 */
	private boolean readCyclic = false;
	
	/**
	 * 循环读取时,信息的处理类
	 */
	private AbstractMessageHandler handler = null;
	
	/**
	 * 循环读取信息的任务
	 */
	private Thread cyclicReadTask = null;
	
	/**
	 * 最近一次操作socket InputStream或者OutputStream的时间
	 */
	private long lastReadTimeMillis = 0;
	
	/**
	 * 实际处理流量
	 */
	private int curFlux = 0;
	
	/**
	 * 上一秒的流量
	 */
	private int lastFlux = 0;
	
	/**
	 * 上一次流量控制开始时间
	 */
	private long lastTime = 0;
	
	public ExtSocket() {
		super();
	}

	public ExtSocket(String host, int port) throws UnknownHostException, IOException {
		super(host, port);
	}

	public boolean isReadCyclic() {
		return readCyclic;
	}

	/**
	 * 是否循环读取信息
	 * @param readCyclic
	 */
	public void setReadCyclic(boolean readCyclic) {
		
		this.readCyclic = readCyclic;
		
		if (this.readCyclic == true) {
			
			if (cyclicReadTask == null || !cyclicReadTask.isAlive()) {
				
				cyclicReadTask = new Thread(new Runnable() {
					
					@Override
					public void run() {
						
						MessageDispacher dispacher = new DefaultMessageDispacher();
						
						while (ExtSocket.this.isReadCyclic() && !ExtSocket.this.isClosed()) {
							
							try {
								
								Message message = ExtSocket.this.read();
								
								if (handler == null) {
									handler = new DefaultMessageHandler();
								}
								
								fluxControll(message, ExtSocket.this);
								
								dispacher.dispach(message, handler, ExtSocket.this);
								
							} catch (IOException e) {
								
								if (e instanceof SocketException) {
									if (ExtSocket.this.isClosed()) {
										logger.info("Socket连接关闭"+ (accountInfo != null ? accountInfo.toString():""));
									} else {
										try {
											ExtSocket.this.close();
										} catch (IOException e1) {
											logger.error(e.getMessage(), e);
										}
									}
								} else {
									logger.error(e.getMessage(), e);
								}
							}
						}
					}
				});
				
				cyclicReadTask.start();
			}
		} else {
			
			if (cyclicReadTask != null) {
				cyclicReadTask.interrupt();
				cyclicReadTask = null;
			}
		}
	}

	public AccountInfo getAccountInfo() {
		return accountInfo;
	}

	public void setAccountInfo(AccountInfo accountInfo) {
		this.accountInfo = accountInfo;
	}

	public long getLastReadTimeMillis() {
		return lastReadTimeMillis;
	}

	public void setLastReadTimeMillis(long lastReadTimeMillis) {
		this.lastReadTimeMillis = lastReadTimeMillis;
	}
	
	/**
	 * 写响应消息
	 * @param message
	 * @param errorCode
	 * @param isProtected
	 * @return
	 */
	public boolean response(Message message, int errorCode) {
		
		boolean respResult = false;
		
		try {
			
			Head head = (Head) message.getHead();
			
			long commandID = head.getCommandID();
			
			Class<? extends Resp> clazz = CommandID.getRespClass(commandID);
			
			if (clazz != null) {
				
				Resp resp = clazz.newInstance();
				resp.setResult(errorCode);
				
				if (resp instanceof Submit2Resp) {
					
					Submit2Resp  submit2Resp = (Submit2Resp) resp;
					
					Submit2 submit2 = (Submit2) message.getBody();
					
					submit2Resp.setSrvNode(submit2.getSrvNode());
					submit2Resp.setServiceCode(submit2.getServiceCode());
					submit2Resp.setSendPort(submit2.getSendPort());
					submit2Resp.setUserNumber(submit2.getUserNumber());
					submit2Resp.setOrgSrvNode(submit2.getOrgSrvNode());
					submit2Resp.setOrgSerialNo(submit2.getOrgSerialNo());
				}
				
				Head respHead = Head.build(resp.length(), CommandID.getRespID(commandID), head.getSerialNoPart1(), head.getSerialNoPart2(), head.getSerialNoPart3());
				
				this.write(new Message(respHead, resp));
				
				respResult = true;
			}
			
		} catch (Exception e) {
			logger.error("Response error : " + e.getMessage());
		}
		
		return respResult;
	}
	
	/**
	 * 写消息
	 * @param message
	 * @throws IOException
	 */
	public void write(Message message) throws IOException {
		
		if (message == null) throw new NullPointerException("Message can not be null !");
		
		synchronized (writeLock) {
			SocketUtils.write(message.toByteArray(), super.getOutputStream());
		}
		
		setLastReadTimeMillis(System.currentTimeMillis());
	}
	
	/**
	 * 读消息
	 * @return
	 * @throws IOException
	 */
	public Message read() throws IOException {
		
		Message message = null;
		
		synchronized (readLock) {
			message = SocketUtils.read(super.getInputStream());
		}
		
		setLastReadTimeMillis(System.currentTimeMillis());
		
		return message;
	}

	public AbstractMessageHandler getHandler() {
		return handler;
	}

	public void setHandler(AbstractMessageHandler messageHandler) {
		this.handler = messageHandler;
	}

	/**
	 * 禁止外部直接使用OutputStream写信息
	 */
	@Deprecated
	public OutputStream getOutputStream() throws IOException {
		throw new UnsupportedOperationException("Please use write(byte[] message) method to send Messages !");
	}

	/**
	 * 禁止外部直接使用InputStream读信息
	 */
	@Deprecated
	public InputStream getInputStream() throws IOException {
		throw new UnsupportedOperationException("Please use read() method to get your Messages !");
	}

	@Override
	public synchronized void close() throws IOException {
		super.close();
		this.setReadCyclic(false);
	}
	
	/**
	 * @param msgHead 接收的消息头
	 * @param clazz 响应信息的body类
	 * @param commandID 命令ID
	 * @param socketHolder 通道连接
	 * @return true 通道达到了最大流量限制, false 通道流量还没有达到最大流量限制
	 */
	public synchronized void fluxControll(Message message, ExtSocket socket) {
		
		if (message == null) {
			logger.warn("fluxControll(Message message, ExtSocket socket)的message实参为空!");
			return;
		}
		
		IBody body = message.getBody();
		
		if (body != null) {
			if (!(body instanceof Submit) && !(body instanceof Submit2)) {
				return;
			}
		} else {
			logger.warn("fluxControll(Message message, ExtSocket socket)的message.getBody()为空!");
			return;
		}
		
		if (socket == null) {
			logger.warn("fluxControll(Message message, ExtSocket socket)的socket实参为空!");
			return;
		}
		
		AccountInfo accountInfo = socket.getAccountInfo();
		
		if (accountInfo == null) {
			logger.warn("fluxControll(Message message, ExtSocket socket)的socket.getAccountInfo()为空!");
			return;
		}
		
		if (accountInfo.isControlFlux() && accountInfo.isLogin()) {  // 流量控制
			
			if (lastTime == 0) {
				lastTime = socket.getLastReadTimeMillis();
			}
			
			curFlux ++;
			
			long interval = socket.getLastReadTimeMillis() - lastTime;
			
			if (interval < 1000) {
				if (curFlux >= accountInfo.getMaxFlux()) {
					logger.debug("业务端" + accountInfo.getSrvNode() + "当前流量:" + curFlux + "(条/秒), 最大流量:" + accountInfo.getMaxFlux() + "(条/秒)");
					lastFlux = curFlux;
					curFlux = 0;
					lastTime = 0;
					try {
						TimeUnit.MILLISECONDS.sleep(1000 - interval);
					} catch (InterruptedException e) {}
				}
			} else {
				logger.debug("业务端" + accountInfo.getSrvNode() + "当前流量:" + curFlux + "(条/秒), 最大流量:" + accountInfo.getMaxFlux() + "(条/秒)");
				lastFlux = curFlux;
				curFlux = 0;
				lastTime = 0;
			}
		}
	}

	public int getLastFlux() {
		return lastFlux;
	}
}
